package com.cctc.rds.scheduler.server.storage.leveldb.manager;

import com.cctc.rds.nrpc.core.util.JsonUtils;
import com.cctc.rds.scheduler.core.task.TaskLog;
import com.cctc.rds.scheduler.core.task.TaskStatus;
import com.cctc.rds.scheduler.server.raft.helper.RaftHelper;
import com.cctc.rds.scheduler.server.storage.leveldb.LevelDB;
import org.iq80.leveldb.DB;
import org.iq80.leveldb.DBIterator;
import org.iq80.leveldb.Snapshot;

import java.io.Closeable;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * LevelDB 内存数据库 - 恢复锁定任务（守护）管理线程。
 * 描述信息：对于 Raft 集群：
 * - 当 Leader 节点挂掉时，对于已经提交到任务调度器服务端的任务，或者没有任务调度器客户端拉取，或者已经被任务调度器客户端拉取但未提交执行结果的任务，都会被
 * 新的 Leader 节点锁定一段时间，然后重置为 pending 状态，重新调度，延迟执行。
 * 注意：如果当前时间已经超过下次延迟执行时间的 15 分钟，视该任务为调度失败，然后重置为 pending 状态，重新调度，延迟执行。。
 */
public class RecoveryLockTaskWorker extends Thread implements Closeable {
    public RecoveryLockTaskWorker() {
        super("recovery-lock-task-worker");
        setDaemon(true);
    }

    @Override
    public void run() {
        while (!Thread.interrupted()) {
            try {
                RaftHelper.checkLeader();
            } catch (Throwable e) {
                try {
                    TimeUnit.SECONDS.sleep(3);
                } catch (InterruptedException ex) {
                }
                continue;
            }
            DB database = LevelDB.get();
            if (database == null) return;
            long curTime = System.currentTimeMillis();
            try (Snapshot snapshot = database.getSnapshot(); DBIterator iterator = database.iterator()) {
                String key = TaskStatus.SCHEDULING.getStatus() + "::";
                iterator.seek(key.getBytes(StandardCharsets.UTF_8));
                while (iterator.hasNext()) {
                    Map.Entry<byte[], byte[]> entry = iterator.next();
                    String value = new String(entry.getValue(), StandardCharsets.UTF_8);
                    TaskLog taskLog = JsonUtils.fromJson(value, TaskLog.class);
                    long scheduleTime = taskLog.getRetryScheduleTime() > 0 ?
                            taskLog.getRetryScheduleTime() : taskLog.getScheduleTime();
                    long nextExecTime = scheduleTime + taskLog.getExecPriod().get(taskLog.getExecPriodIndex());
                    if (nextExecTime - curTime < 15 * 60) continue;
                    try {
                        RaftHelper.checkLeader();
                        taskLog.setStatus(TaskStatus.PENDING.getStatus());
                        database.put(entry.getKey(), JsonUtils.toJson(taskLog).getBytes(StandardCharsets.UTF_8));
                    } catch (Exception e) {
                        break;
                    }
                }
            } catch (IOException e) {
                return;
            }
            try {
                TimeUnit.SECONDS.sleep(6);
            } catch (InterruptedException e) {
            }
        }
    }

    @Override
    public void close() throws IOException {
        this.interrupt();
    }
}
